package il.technion.ewolf.beehive.handlers;

import java.io.Serializable;
import java.util.Arrays;
import java.util.Collection;
import java.util.LinkedList;
import java.util.concurrent.atomic.AtomicInteger;

import com.google.inject.Inject;
import com.google.inject.Provider;
import com.google.inject.name.Named;

import il.technion.ewolf.beehive.cache.BeeHiveStorage;
import il.technion.ewolf.beehive.cache.DataCollector;
import il.technion.ewolf.beehive.op.MyFindValueOperation;
import il.technion.ewolf.kbr.openkad.msg.ReplicationRequest;
import il.technion.ewolf.kbr.Key;
import il.technion.ewolf.kbr.Node;
import il.technion.ewolf.kbr.openkad.handlers.AbstractHandler;
import il.technion.ewolf.kbr.openkad.msg.KadMessage;
import il.technion.ewolf.kbr.openkad.net.MessageDispatcher;
import il.technion.ewolf.kbr.openkad.net.filter.IdMessageFilter;
import il.technion.ewolf.kbr.openkad.net.filter.MessageFilter;
import il.technion.ewolf.kbr.openkad.net.filter.TypeMessageFilter;

public class BeehiveReplicationMessageHandler extends AbstractHandler{

	Provider<MessageDispatcher<Void>> msgDispatcherProvider;
	BeeHiveStorage storage;
	DataCollector dataCollector;
	Provider<MyFindValueOperation> myFindValueOperationProvider;
	private Node localNode;
	private final Provider<ReplicationRequest> replicationRequestProvider;
	private AtomicInteger nrToldToDelete;
	private AtomicInteger nrNumberToldToDelete;
	private AtomicInteger nrNumberOfMsg;
	
	@Inject
	BeehiveReplicationMessageHandler(
			Provider<MessageDispatcher<Void>> msgDispatcherProvider,
			BeeHiveStorage storage,
			DataCollector dataCollector,
			Provider<MyFindValueOperation> myFindValueOperationProvider,
			@Named("openkad.local.node") Node localNode,
			@Named("openkad.testing.nToldToDelete") AtomicInteger toldToDelete,
			@Named("openkad.testing.nNumberToldToDeleteMsg") AtomicInteger numberToldToDelete,
			@Named("openkad.testing.nRepMessages") AtomicInteger numberOfMsg,
			Provider<ReplicationRequest> replicationRequestProvider) {
		
		super(msgDispatcherProvider);
		this.msgDispatcherProvider = msgDispatcherProvider;
		this.storage = storage;
		this.dataCollector = dataCollector;
		this.myFindValueOperationProvider = myFindValueOperationProvider;
		this.localNode = localNode;
		this.replicationRequestProvider = replicationRequestProvider;
		this.nrToldToDelete = toldToDelete;
		this.nrNumberToldToDelete = numberToldToDelete;
		this.nrNumberOfMsg = numberOfMsg;
	}

	@Override
	public void completed(KadMessage msg, Void attachment) {
		ReplicationRequest replicationMsg = ((ReplicationRequest)msg);
		
		LinkedList<Key> objectList = replicationMsg.getObjectList();
		Boolean toReplicate = replicationMsg.getToReplicate();
		
		if (toReplicate){
			synchronized (storage.getMutex()) {
				//when needs to replicate
				LinkedList<Key> notToReplicate = new LinkedList<Key>();
				Key myKey = localNode.getKey();
				
				for (Key objectKey: objectList){
					Integer myLevelFromObject = objectKey.getBitLength() - objectKey.xor(myKey).getFirstSetBitIndex();
					Integer levelShouldBe = dataCollector.getLevelOfKey(objectKey);
					if ( dataCollector.getObjectLevelsListSize() != 0){
						if ((levelShouldBe == null) || ((levelShouldBe != null) && (myLevelFromObject < levelShouldBe))){ 
							notToReplicate.add(objectKey);
						} else {
							//replication
							replicate(objectKey);
						}
					}
				}
				//adding objects that are not replicated any more
				notToReplicate.addAll(storage.getStorageDiff());
				if (notToReplicate.size() != 0){
					sendReplicationRequests(notToReplicate, myKey, replicationMsg.getSrcNode(), false);
				}
				//System.out.println("End dealing with objects to replication, " + localNode.getKey().toString());
			}
		} else{
			synchronized (storage.getMutex()) {
				//when needs to delete from storage
				nrNumberOfMsg.incrementAndGet();
				nrToldToDelete.addAndGet(objectList.size());
				nrNumberToldToDelete.incrementAndGet();
				for (Key objectKey : objectList){
					storage.remove(objectKey);
				}
				//System.out.println("Finished deleting objects, " + localNode.getKey().toString());
			}
		}
	}

	private void sendReplicationRequests(LinkedList<Key> objectList, Key localNodeKey, Node nodeToGetReplications, Boolean toReplicate){
		ReplicationRequest msg = replicationRequestProvider.get()
		.setObjectList(objectList)
		.setToReplicate(toReplicate);
	
		msgDispatcherProvider.get()
		.addFilter(new IdMessageFilter(msg.getId()))
		.addFilter(new TypeMessageFilter(ReplicationRequest.class))
		.setConsumable(false)
		.send(nodeToGetReplications, msg);
	}
	
	private void replicate(Key key){
		//get the object and insert it to the storage
		//maybe find the object in storage
		Serializable theObject = storage.search(key);
		if (theObject == null){
    		MyFindValueOperation op = myFindValueOperationProvider.get()
    		.setKey(key)
    		.setReplication(true);
    		theObject = op.doFindValue();
		}
		storage.insert(key, theObject);
	}
	
	@Override
	public void failed(Throwable exc, Void attachment) {
		System.out.println("Exception: " + exc.getLocalizedMessage() + " happened");
		System.out.println("Stack trace: ");
		exc.printStackTrace();
	}

	@Override
	protected Collection<MessageFilter> getFilters() {
		// only accept ReplicationRequests messages
		return Arrays.asList(new MessageFilter[] {
				new TypeMessageFilter(ReplicationRequest.class)
		});
	}
}
